package com.lwq.im.chartdemo.handler;

import cn.hutool.core.util.StrUtil;
import com.lwq.im.chartdemo.bean.InitBean;
import com.lwq.im.chartdemo.bean.Ping;
import com.lwq.im.chartdemo.bean.RequestBody;
import com.lwq.im.chartdemo.constants.Constants;
import com.lwq.im.chartdemo.endecode.MessageDecoder;
import com.lwq.im.chartdemo.endecode.MessageEncoder;
import com.lwq.im.chartdemo.util.SystemUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * socket 处理器
 */
@ChannelHandler.Sharable
@Slf4j
public class SocketChartHandler extends SimpleChannelInboundHandler<RequestBody> {



    /**
     *  默认的写空闲时间(秒)
     */
    public final Long DEFAULT_WRITE_IDLE = 30L;

    /**
     *  默认的读空闲时间(秒)
     */
    public final Long DEFAULT_READ_IDLE = 30L;

    /**
     * 最大请求无响应次数
     */
    private final Integer MAX_PING_COUNT = 5;

    /**
     * websocket 端口号
     */
    private final Integer port;

    /**
     * 请求处理器
     */
    private final ChartRequestHandler chartRequestHandler;

    /**
     *  写空闲时间(秒)
     */
    public Long writeIdle;

    /**
     *  读空闲时间(秒)
     */
    public Long readIdle;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;


    private ServerBootstrap createServerBootstrap(){
        if (SystemUtil.linuxSystem()){
            bossGroup = new EpollEventLoopGroup();
            workerGroup = new EpollEventLoopGroup();
        }else {
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
        }
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.channel(SystemUtil.linuxSystem() ? EpollServerSocketChannel.class : NioServerSocketChannel.class);
        return bootstrap;
    }

    public void start(){
        doStart();
    }

    private void doStart() {
        ServerBootstrap bootstrap = createServerBootstrap();
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch){
                // http 的编码解密器
                ch.pipeline().addLast(new HttpServerCodec());
                ch.pipeline().addLast(new ChunkedWriteHandler());
                // http 在传输中是分段的， HttpObjectAggregator 可以将多个段 聚合
                ch.pipeline().addLast(new HttpObjectAggregator(65536));
                ch.pipeline().addLast(new WebSocketServerProtocolHandler("/"));
                ch.pipeline().addLast(new MessageDecoder());
                ch.pipeline().addLast(new MessageEncoder());
                // 心跳检查处理器
                ch.pipeline().addLast(new IdleStateHandler(readIdle, writeIdle, 0, TimeUnit.SECONDS));
                ch.pipeline().addLast(SocketChartHandler.this);
            }

        });

        ChannelFuture channelFuture = bootstrap.bind(port).syncUninterruptibly();
        channelFuture.channel().newSucceededFuture().addListener(future -> {
            String logMessage = "\r\n"
                    +"======================================================\r\n" +
                    "websocket 服务启动。端口:{}\r\n" +
                    "=======================================================\r\n";
            log.info(logMessage, port);
        });
        channelFuture.channel().closeFuture().addListener(future -> this.destroy(bossGroup,workerGroup));
    }

    public void destroy(EventLoopGroup bossGroup , EventLoopGroup workerGroup) {
        if(bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isShutdown() ) {
            try {
                bossGroup.shutdownGracefully();
            }catch(Exception e) {
                // 静默处理
            }
        }

        if(workerGroup != null && !workerGroup.isShuttingDown() && !workerGroup.isShutdown() ) {
            try {
                workerGroup.shutdownGracefully();
            }catch(Exception e) {
                // 静默处理
            }
        }
    }

    public void destroy() {
        this.destroy(bossGroup,workerGroup);
    }

    /**
     * 构造方法
     *
     * @param initBean
     */
    public SocketChartHandler(InitBean initBean) {
        this.port = initBean.getPort();
        this.chartRequestHandler = initBean.getChartRequestHandler();
        if (null == initBean.getReadIdle()){
            initBean.setReadIdle(DEFAULT_READ_IDLE);
        }
        if (null == initBean.getWriteIdle()){
            initBean.setWriteIdle(DEFAULT_WRITE_IDLE);
        }
        this.readIdle = initBean.getReadIdle();
        this.writeIdle = initBean.getWriteIdle();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RequestBody requestBody) throws Exception {
        chartRequestHandler.doHandle(channelHandlerContext.channel(), requestBody);
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        log.info("通道{}注册", ctx.channel().id().asShortText());
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        log.info("通道{}未注册", ctx.channel().id().asShortText());
        super.channelUnregistered(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("通道{}激活", ctx.channel().id().asShortText());
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("通道{}关闭", ctx.channel().id().asShortText());
        String userId = ctx.channel().attr(Constants.ChannelAttrKye.USER_ID).get();
        if (StrUtil.isBlank(userId)){
            return;
        }
        log.info("执行关闭通道{}", ctx.channel().id().asShortText());
        RequestBody requestBody = new RequestBody();
        requestBody.setOpType(Constants.OpType.OP_TYPE_CLOSE);
        chartRequestHandler.doHandle(ctx.channel(), requestBody);

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }


    /**
     * 处理心跳 IdleStateEvent 是心跳事件，通过 IdleStateHandler 来注册进来的。
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (!(evt instanceof IdleStateEvent)){
            return;
        }

        String userId = ctx.channel().attr(Constants.ChannelAttrKye.USER_ID).get();

        if (((IdleStateEvent) evt).state().equals(IdleState.WRITER_IDLE) && StrUtil.isBlank(userId)){
            ctx.close();
        }
        
        AttributeKey<Integer> attributePingKey = Constants.ChannelAttrKye.PING_COUNT;
        if (((IdleStateEvent) evt).state().equals(IdleState.WRITER_IDLE) && StrUtil.isNotBlank(userId)){
            Integer pingCount = ctx.channel().attr(attributePingKey).get();
            ctx.channel().attr(attributePingKey).set(null == pingCount ? 1 : pingCount + 1);
            ctx.channel().writeAndFlush(Ping.getInstance());
            return;
        }

        Integer pingCount = ctx.channel().attr(attributePingKey).get();
        if (((IdleStateEvent) evt).state().equals(IdleState.READER_IDLE) && pingCount != null && pingCount >= MAX_PING_COUNT ){
            ctx.channel().close();
            log.info("通道{}请求超过了{}次无响应，关闭通道",ctx.channel().id().asShortText(),MAX_PING_COUNT);
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("出现异常",cause);
        super.exceptionCaught(ctx, cause);
    }
}
